/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.split.reducers;

import aiaudio.processing.base.mapreduce.CoordinateDoubleValueKey;
import aiaudio.Application;
import aiaudio.database.tables.ArtistsListTable;
import aiaudio.database.tables.RatingsMatrixTable;
import java.io.IOException;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;

/**
 *
 * @author nastya
 */
public class DataSetSplitReducer extends TableReducer<ImmutableBytesWritable, CoordinateDoubleValueKey, ImmutableBytesWritable> {

    public static final String SPLITTING_COEFFICIENT_PARAM = "splittingCoefficient";
    public static final String ARTISTS_TABLE_NAME_PARAMETER = "artistsTableName";
    private double splittingCoefficient;
    HTable artistsTable;

    @Override
    protected void reduce(ImmutableBytesWritable key, Iterable<CoordinateDoubleValueKey> values, Context context) throws IOException, InterruptedException {

        byte[] artistId = key.get();
        if(RatingsMatrixTable.isDefaultValue(artistId)){
            return;
        }
        long fullVotingCount = getVotingCountByArtist(artistId);
        long usersToSave = (long) ((double) fullVotingCount * splittingCoefficient);
        if (usersToSave <= 0) {
            usersToSave = 1;
        }

        long c = 0;
        Put put = new Put(artistId);
        for (CoordinateDoubleValueKey userValueKey : values) {
            if (c >= usersToSave) {
                break;
            }
            RatingsMatrixTable.addUserRating(put, userValueKey.getCoordinate(), userValueKey.getValue());

            c++;
        }
        context.write(new ImmutableBytesWritable(key.get()), put);

    }

    protected long getVotingCountByArtist(byte[] key) throws IOException {
        return ArtistsListTable.getVotingCount(artistsTable, key);
    }

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        super.setup(context);
        splittingCoefficient = context.getConfiguration().getFloat(SPLITTING_COEFFICIENT_PARAM, 0.5f);
        String usersTableName = context.getConfiguration().get(ARTISTS_TABLE_NAME_PARAMETER);
        artistsTable = Application.database().getByName(usersTableName).getTable();
    }
}
